package com.ecommerce.analysis

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._

class HotKeywordAnalyzer extends BaseAnalyzer {
  override def analyze(spark: SparkSession, cleanedData: DataFrame): DataFrame = {
    logger.info("===== 开始执行热门关键词分析 =====")
    try {
      val result = cleanedData
        .select(explode(split(lower(col("Description")), " ")).alias("keyword"))
        .filter(
          length(col("keyword")) >= 3 &&
            not(col("keyword").isin("the", "and", "of", "for", "with"))
        )
        .groupBy("keyword")
        .agg(count("*").alias("keyword_count"))
        .orderBy(col("keyword_count").desc)
        .limit(50)

      logger.info(s"热门关键词分析完成，结果共 ${result.count()} 条记录")
      result.show(5, truncate = false)
      result
    } catch {
      case e: Exception =>
        logger.error("热门关键词分析失败", e)
        throw e
    }
  }

  override def performSave(result: DataFrame, tableName: String, spark: SparkSession): Unit = {
    createTableIfNotExists(result, tableName, "keyword")
    super.performSave(result, tableName, spark)
  }
}